package com.cnsugar.common.hbase;

import com.cnsugar.common.hbase.entity.IHBaseEntity;
import com.cnsugar.common.hbase.utils.TypeUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * HBase操作工具类
 *
 * @Author Sugar
 * @Version 2018/5/31 18:42
 */
public class HBaseUtils {
    private static Logger logger = LoggerFactory.getLogger(HBaseUtils.class);
    public static final String FAMILY = "f1";//默认列簇名
    public static final byte[] FAMILY_BYTES = Bytes.toBytes(FAMILY);
    public static final String ROW = "ROW";

    private static Connection conn;

    static {
        Configuration conf = HBaseConfiguration.create();
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }

        logger.info("hbase.zookeeper.quorum: " + conf.get("hbase.zookeeper.quorum"));
    }

    public static HTable getTable(String table) throws IOException {
        return (HTable) conn.getTable(TableName.valueOf(table));
    }

    /**
     * 创建表
     *
     * @param table 表名
     */
    public static boolean createTable(String table) {
        return createTable(table, FAMILY);
    }

    /**
     * 创建表
     *
     * @param table    表名
     * @param families 列族
     */
    public static boolean createTable(String table, String... families) {
        TableName tableName = TableName.valueOf(table);
        Admin admin = null;
        try {
            admin = conn.getAdmin();
            if (admin.tableExists(tableName)) {
                logger.warn("Table already exists in HBase: {}", tableName);
                return false;
            } else {
                HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
                for (String family : families) {
                    tableDescriptor.addFamily(new HColumnDescriptor(family));
                }
                admin.createTable(tableDescriptor);
            }
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (admin != null) {
                try {
                    admin.close();
                } catch (IOException e) {
                }
            }
        }
        return false;
    }

    /**
     * 通过rowkey查询单条记录，不限定列簇
     *
     * @param rowkey
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T get(String rowkey, Class<T> clazz) {
        String table = TypeUtils.getHTableName(clazz);
        return get(table, null, rowkey, clazz);
    }

    /**
     * 通过rowkey查询单条记录，不限定列簇
     *
     * @param table
     * @param rowkey
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T get(String table, String rowkey, Class<T> clazz) {
        return get(table, null, rowkey, clazz);
    }

    /**
     * 通过rowkey查询单条记录
     *
     * @param table
     * @param rowkey
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T get(String table, String family, String rowkey, Class<T> clazz) {
        HTable htable = null;
        try {
            htable = getTable(table);
            Result result = htable.get(new Get(Bytes.toBytes(rowkey)));
            return TypeUtils.resultToBean(result, family, clazz);
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
        return null;
    }

    /**
     * 通过rowkey查询多条记录，不限定列簇
     *
     * @param clazz
     * @param rowkeys
     * @param <T>
     * @return
     */
    public static <T> List<T> get(Class<T> clazz, String... rowkeys) {
        String table = TypeUtils.getHTableName(clazz);
        return get(table, null, clazz, rowkeys);
    }

    /**
     * 通过rowkey查询多条记录，不限定列簇
     *
     * @param table
     * @param clazz
     * @param rowkeys
     * @param <T>
     * @return
     */
    public static <T> List<T> get(String table, Class<T> clazz, String... rowkeys) {
        return get(table, null, clazz, rowkeys);
    }

    /**
     * 通过rowkey查询多条记录
     *
     * @param table
     * @param family
     * @param clazz
     * @param rowkeys
     * @param <T>
     * @return
     */
    public static <T> List<T> get(String table, String family, Class<T> clazz, String... rowkeys) {
        HTable htable = null;
        try {
            htable = getTable(table);
            List<T> list = new ArrayList<>(rowkeys.length);
            if (family == null) {//在循环外判断，可以减少if判断次数，以提高性能
                for (String rowkey : rowkeys) {
                    Result result = htable.get(new Get(Bytes.toBytes(rowkey)));
                    T obj = TypeUtils.resultToBean(result, clazz);
                    if (obj == null) {
                        continue;
                    }
                    list.add(obj);
                }
            } else {
                byte[] familyBytes = Bytes.toBytes(family);
                for (String rowkey : rowkeys) {
                    Result result = htable.get(new Get(Bytes.toBytes(rowkey)));
                    T obj = TypeUtils.resultToBean(result, familyBytes, clazz);
                    if (obj == null) {
                        continue;
                    }
                    list.add(obj);
                }
            }
            return list;
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
        return null;
    }

    /**
     * 通过rowkey查询单条记录，不限制列簇,Value通过实现IValueMapper接口进行自由处理
     * 适用于只有一个family
     *
     * @param table
     * @param rowkey
     * @param valueMapper
     * @return
     */
    public static Map<String, Object> get(String table, String rowkey, IValueMapper valueMapper) {
        return get(table, null, rowkey, valueMapper);
    }

    /**
     * 通过rowkey查询单条记录,Value通过实现IValueMapper接口进行自由处理
     * 适用于有多个family
     *
     * @param table
     * @param family
     * @param rowkey
     * @param valueMapper
     * @return
     */
    public static Map<String, Object> get(String table, String family, String rowkey, IValueMapper valueMapper) {
        HTable htable = null;
        try {
            htable = getTable(table);
            Result result = htable.get(new Get(Bytes.toBytes(rowkey)));
            return TypeUtils.resultToMap(result, family, valueMapper);
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
        return null;
    }

    /**
     * 通过rowkey查询多条记录
     *
     * @param table
     * @param valueMapper
     * @param rowkeys
     * @return
     */
    public static List<Map<String, Object>> get(String table, IValueMapper valueMapper, String... rowkeys) {
        return get(table, null, valueMapper, rowkeys);
    }

    /**
     * 通过rowkey查询多条记录
     *
     * @param table
     * @param valueMapper
     * @param rowkeys
     * @return
     */
    public static List<Map<String, Object>> get(String table, String family, IValueMapper valueMapper, String... rowkeys) {
        HTable htable = null;
        try {
            htable = getTable(table);
            List<Map<String, Object>> list = new ArrayList<>(rowkeys.length);
            if (family == null) {//在循环外判断，可以减少if判断次数，以提高性能
                for (String rowkey : rowkeys) {
                    Result result = htable.get(new Get(Bytes.toBytes(rowkey)));
                    Map<String, Object> map = TypeUtils.resultToMap(result, valueMapper);
                    if (map == null) {
                        continue;
                    }
                    list.add(map);
                }
            } else {
                byte[] familyBytes = Bytes.toBytes(family);
                for (String rowkey : rowkeys) {
                    Result result = htable.get(new Get(Bytes.toBytes(rowkey)));
                    Map<String, Object> map = TypeUtils.resultToMap(result, familyBytes, valueMapper);
                    if (map == null) {
                        continue;
                    }
                    list.add(map);
                }
            }

            return list;
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
        return null;
    }

    /**
     * scan获取数据，不限定列簇
     *
     * @param scan
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> List<T> scan(Scan scan, Class<T> clazz) {
        String table = TypeUtils.getHTableName(clazz);
        return scan(table, null, scan, clazz);
    }

    /**
     * scan获取数据，指定列簇
     *
     * @param scan
     * @param family
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> List<T> scan(Scan scan, String family, Class<T> clazz) {
        String table = TypeUtils.getHTableName(clazz);
        return scan(table, family, scan, clazz);
    }

    /**
     * scan获取数据，不限定列簇
     *
     * @param table
     * @param scan
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> List<T> scan(String table, Scan scan, Class<T> clazz) {
        return scan(table, null, scan, clazz);
    }

    /**
     * scan获取数据
     *
     * @param table
     * @param family
     * @param scan
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> List<T> scan(String table, String family, Scan scan, Class<T> clazz) {
        HTable htable = null;
        ResultScanner scanner = null;
        try {
            htable = getTable(table);
            List<T> list = new ArrayList<>();
            scanner = htable.getScanner(scan);
            if (family == null) {//在循环外判断，可以减少if判断次数，以提高性能
                for (Result result : scanner) {
                    T obj = TypeUtils.resultToBean(result, clazz);
                    if (obj == null) {
                        continue;
                    }
                    list.add(obj);
                }
            } else {
                byte[] familyBytes = Bytes.toBytes(family);
                for (Result result : scanner) {
                    T obj = TypeUtils.resultToBean(result, familyBytes, clazz);
                    if (obj == null) {
                        continue;
                    }
                    list.add(obj);
                }
            }
            return list;
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (scanner != null) {
                scanner.close();
            }
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
        return null;
    }

    /**
     * scan获取数据，不限制列簇
     *
     * @param table
     * @param scan
     * @param valueMapper
     * @return
     */
    public static List<Map<String, Object>> scan(String table, Scan scan, IValueMapper valueMapper) {
        return scan(table, null, scan, valueMapper);
    }

    /**
     * scan获取数据
     *
     * @param table
     * @param family
     * @param scan
     * @param valueMapper
     * @return
     */
    public static List<Map<String, Object>> scan(String table, String family, Scan scan, IValueMapper valueMapper) {
        HTable htable = null;
        ResultScanner scanner = null;
        try {
            htable = getTable(table);
            List<Map<String, Object>> list = new ArrayList<>();
            scanner = htable.getScanner(scan);
            if (family == null) {//在循环外判断，可以减少if判断次数，以提高性能
                for (Result result : scanner) {
                    Map<String, Object> map = TypeUtils.resultToMap(result, valueMapper);
                    if (map == null) {
                        continue;
                    }
                    list.add(map);
                }
            } else {
                byte[] familyBytes = Bytes.toBytes(family);
                for (Result result : scanner) {
                    Map<String, Object> map = TypeUtils.resultToMap(result, familyBytes, valueMapper);
                    if (map == null) {
                        continue;
                    }
                    list.add(map);
                }
            }
            return list;
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (scanner != null) {
                scanner.close();
            }
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
        return null;
    }

    /**
     * 保存或新增数据
     *
     * @param entity
     */
    public static void put(IHBaseEntity entity) {
        put(entity, FAMILY_BYTES);
    }

    /**
     * 保存或新增数据
     *
     * @param entity
     * @param family
     */
    public static void put(IHBaseEntity entity, byte[] family) {
        String table = TypeUtils.getHTableName(entity.getClass());
        put(entity, table, family);
    }

    /**
     * 保存或新增数据
     *
     * @param entity
     * @param table
     */
    public static void put(IHBaseEntity entity, String table) {
        put(entity, table, FAMILY_BYTES);
    }

    /**
     * 保存或新增数据
     *
     * @param entity
     * @param table
     * @param family
     */
    public static void put(IHBaseEntity entity, String table, byte[] family) {
        if (entity.getRow() == null) {
            logger.error("Row key is null: {}", entity);
            return;
        }
        HTable htable = null;
        try {
            htable = getTable(table);
            Put put = TypeUtils.beanToPut(entity, family);
            htable.put(put);
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
    }

    /**
     * 批量保存或新增数据
     *
     * @param list
     */
    public static void put(List<IHBaseEntity> list) {
        put(list, FAMILY_BYTES);
    }

    /**
     * 批量保存或新增数据
     *
     * @param list
     * @param family
     */
    public static void put(List<IHBaseEntity> list, byte[] family) {
        if (list == null || list.isEmpty()) {
            return;
        }
        String table = TypeUtils.getHTableName(list.get(0).getClass());
        put(list, table, family);
    }

    /**
     * 批量保存或新增数据
     *
     * @param list
     * @param table
     */
    public static void put(List<IHBaseEntity> list, String table) {
        put(list, table, FAMILY_BYTES);
    }

    /**
     * 批量保存或新增数据
     *
     * @param list
     * @param table
     * @param family
     */
    public static void put(List<IHBaseEntity> list, String table, byte[] family) {
        HTable htable = null;
        try {
            htable = getTable(table);
            List<Put> putList = new ArrayList<>(list.size());
            list.forEach(entity -> {
                if (entity.getRow() == null) {
                    logger.error("Row key is null: {}", entity);
                    return;
                }
                Put put = TypeUtils.beanToPut(entity, family);
                putList.add(put);
            });
            htable.put(putList);
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
    }

    /**
     * 保存或新增数据
     *
     * @param map
     * @param table
     */
    public static void putMap(Map<String, Object> map, String table) {
        putMap(map, table, FAMILY_BYTES);
    }

    /**
     * 保存或新增数据
     *
     * @param map
     * @param row
     * @param table
     */
    public static void putMap(Map<String, Object> map, String row, String table) {
        putMap(map, row, table, FAMILY_BYTES);
    }

    /**
     * 保存或新增数据
     *
     * @param map
     * @param table
     * @param family
     */
    public static void putMap(Map<String, Object> map, String table, byte[] family) {
        String row = (String) map.get(ROW);
        if (StringUtils.isEmpty(row)) {
            logger.error("Row key is null: {}", map);
            return;
        } else {
            map.remove(ROW);//从内容中移除rowkey
        }
        putMap(map, row, table, family);
    }

    /**
     * 保存或新增数据
     *
     * @param map
     * @param row
     * @param table
     * @param family
     */
    public static void putMap(Map<String, Object> map, String row, String table, byte[] family) {
        if (row == null) {
            logger.error("Row key is null: {}", map);
            return;
        }
        HTable htable = null;
        try {
            htable = getTable(table);
            Put put = TypeUtils.mapToPut(map, row, family);
            htable.put(put);
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
    }

    /**
     * 批量保存或新增数据
     *
     * @param list
     * @param table
     */
    public static void putMap(List<Map<String, Object>> list, String table) {
        putMap(list, table, FAMILY_BYTES);
    }

    /**
     * 批量保存或新增数据
     *
     * @param list
     * @param table
     * @param family
     */
    public static void putMap(List<Map<String, Object>> list, String table, byte[] family) {
        HTable htable = null;
        try {
            htable = getTable(table);
            List<Put> putList = new ArrayList<>(list.size());
            list.forEach(map -> {
                String row = (String) map.get(ROW);
                if (StringUtils.isEmpty(row)) {
                    logger.error("Row key is null: {}", map);
                    return;
                } else {
                    map.remove(ROW);//从内容中移除rowkey
                }
                Put put = TypeUtils.mapToPut(map, row, family);
                putList.add(put);
            });
            htable.put(putList);
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
    }

    /**
     * 通过rowkey删除数据
     *
     * @param table
     * @param rowkeys
     */
    public static void delete(String table, String... rowkeys) {
        HTable htable = null;
        try {
            htable = getTable(table);
            if (rowkeys.length == 1) {
                htable.delete(new Delete(Bytes.toBytes(rowkeys[0])));
            } else {
                List<Delete> list = new ArrayList<>(rowkeys.length);
                for (String rowkey : rowkeys) {
                    list.add(new Delete(Bytes.toBytes(rowkey)));
                }
                htable.delete(list);
            }
        } catch (TableNotFoundException e) {
            logger.error("Table not found in HBase: {}", table);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (htable != null) {
                try {
                    htable.close();
                } catch (IOException e) {
                }
            }
        }
    }
}
